6cd7cdf94c7dfddc3602c26bdec3fc401cd66cc4,core/client/src/main/java/alluxio/client/netty/NettyUnderFileSystemFileWriter.java,NettyUnderFileSystemFileWriter,write,#InetSocketAddress#number#number#number[]#number#number#,60

Before Change


      int offset, int length) throws IOException {
    SingleResponseListener listener = null;
    try {
      ChannelFuture f = mClientBootstrap.connect(address).sync();

      LOG.debug("Connected to remote machine {}", address);
      Channel channel = f.channel();
      listener = new SingleResponseListener();
      mHandler.addListener(listener);
      channel.writeAndFlush(new RPCFileWriteRequest(ufsFileId, fileOffset, length,
          new DataByteArrayChannel(source, offset, length)));

      RPCResponse response = listener.get(NettyClient.TIMEOUT_MS, TimeUnit.MILLISECONDS);
      channel.close().sync();

      switch (response.getType()) {
        case RPC_FILE_WRITE_RESPONSE:

After Change


  }

  @Override
  public void write(InetSocketAddress address, long ufsFileId, long fileOffset, byte[] source,
      int offset, int length) throws IOException {
    SingleResponseListener listener = null;
    Channel channel = null;
    Metrics.NETTY_UFS_WRITE_OPS.inc();
    try {
      channel = BlockStoreContext.acquireNettyChannel(address, mClientBootstrap) ;
      listener = new SingleResponseListener();
      ((ClientHandler) channel.pipeline().last()).addListener(listener);
      ChannelFuture channelFuture = channel.writeAndFlush(
          new RPCFileWriteRequest(ufsFileId, fileOffset, length,
              new DataByteArrayChannel(source, offset, length))).sync();

      if (channelFuture.isDone() && !channelFuture.isSuccess()) {
        LOG.error("Failed to read ufs file from %s for ufsFilId %d with error %s.",
            address.toString(), ufsFileId, channelFuture.cause());
        throw new IOException(channelFuture.cause());
      }

      RPCResponse response = listener.get(NettyClient.TIMEOUT_MS, TimeUnit.MILLISECONDS);

      switch (response.getType()) {
        case RPC_FILE_WRITE_RESPONSE:
          RPCFileWriteResponse resp = (RPCFileWriteResponse) response;
          RPCResponse.Status status = resp.getStatus();
          LOG.debug("status: {} from remote machine {} received", status, address);

          if (status != RPCResponse.Status.SUCCESS) {
            throw new IOException(ExceptionMessage.UNDER_FILE_WRITE_ERROR.getMessage(ufsFileId,
                address, status.getMessage()));
          }
          break;
        case RPC_ERROR_RESPONSE:
          RPCErrorResponse error = (RPCErrorResponse) response;
          throw new IOException(error.getStatus().getMessage());
        default:
          throw new IOException(ExceptionMessage.UNEXPECTED_RPC_RESPONSE
              .getMessage(response.getType(), RPCMessage.Type.RPC_FILE_WRITE_RESPONSE));
      }
    } catch (Exception e) {
      Metrics.NETTY_UFS_WRITE_FAILURES.inc();
      throw new IOException(e);
    } finally {
      if (channel != null && listener != null) {